#include "StdAfx.h"
#include "..\include\Thread.h"
#include "..\include\WorkQueue.h"
#include "..\include\LogManager.h"

using namespace Common;

//---------------------------------------------------------------------
WorkQueue::Request::Request(uint16 channel, uint16 rtype, const Any& rData, uint16 retry, RequestID rid)
	: m_uChannel(channel), m_uType(rtype), m_anyData(rData), m_uRetryCount(retry), m_uID(rid), m_bAborted(false)
{

}
//---------------------------------------------------------------------
WorkQueue::Request::~Request()
{

}
//---------------------------------------------------------------------
//---------------------------------------------------------------------
WorkQueue::Response::Response(const Request* rq, bool success, const Any& data, const std::string& msg)
	: m_pRequest(rq), m_bSuccess(success), m_strMessages(msg), m_anyData(data)
{

}
//---------------------------------------------------------------------
WorkQueue::Response::~Response()
{
	if( m_pRequest != NULL )
		delete m_pRequest;
}
//---------------------------------------------------------------------
//---------------------------------------------------------------------
WorkQueue::WorkQueue( const std::string& strWorkQueueName )
	: m_strWorkQueueName(strWorkQueueName)
	, m_uWorkThreadCount(1)
	, m_bWorkerRenderSystemAccess(false)
	, m_bIsRunning(false)
	, m_uResponseTimeLimitMS(8)
	, m_pWorkerFunc(NULL)
	, m_uRequestCount(0)
	, m_bPaused(false)
	, m_bAcceptRequests(true)
{
}
//---------------------------------------------------------------------
WorkQueue::~WorkQueue()
{
	//shutdown(); // can't call here; abstract function

	for (RequestQueue::iterator pos = m_RequestQueue.begin(); pos != m_RequestQueue.end(); ++pos)
	{
		if( *pos != NULL )
			delete *pos;
	}
	m_RequestQueue.clear();

	for (ResponseQueue::iterator pos = m_ResponseQueue.begin(); pos != m_ResponseQueue.end(); ++pos)
	{
		if( *pos != NULL )
			delete *pos;
	}
	m_ResponseQueue.clear();
}
//---------------------------------------------------------------------
uint16 WorkQueue::getChannel(const std::string& strChannelName)
{
	LOCK_MUTEX(m_mtxChannelMapMutex)

	ChannelMap::iterator pos = m_ChannelMap.find(strChannelName);
	if (pos == m_ChannelMap.end())
	{
		pos = m_ChannelMap.insert(ChannelMap::value_type(strChannelName, m_nextChannel++)).first;
	}
	return pos->second;
}
//---------------------------------------------------------------------
const std::string& WorkQueue::getName() const
{
	return m_strWorkQueueName;
}
//---------------------------------------------------------------------
uint16 WorkQueue::getWorkerThreadCount() const
{
	return m_uWorkThreadCount;
}
//---------------------------------------------------------------------
void WorkQueue::setWorkerThreadCount(size_t threadCount)
{
	m_uWorkThreadCount = threadCount;
}
//---------------------------------------------------------------------
bool WorkQueue::getWorkersCanAccessRenderSystem() const
{
	return m_bWorkerRenderSystemAccess;
}
//---------------------------------------------------------------------
void WorkQueue::setWorkersCanAccessRenderSystem(bool access)
{
	m_bWorkerRenderSystemAccess = access;
}

//---------------------------------------------------------------------
void WorkQueue::addRequestHandler(uint16 channel, RequestHandler* rh)
{
	LOCK_RW_MUTEX_WRITE(m_mtxRequestHandlerMutex);

	//search whether the handler has already existed.
	RequestHandlerListByChannel::iterator pos = m_RequestHandlers.find(channel);
	if (pos == m_RequestHandlers.end())
		pos = m_RequestHandlers.insert(RequestHandlerListByChannel::value_type(channel, RequestHandlerList())).first;

	RequestHandlerList& handlers = pos->second;
	bool duplicate = false;
	for (RequestHandlerList::iterator pos = handlers.begin(); pos != handlers.end(); ++pos)
	{
		if ( *pos == rh)
		{
			duplicate = true;
			break;
		}
	}
	if (!duplicate)
		handlers.push_back( rh );
}
//---------------------------------------------------------------------
void WorkQueue::removeRequestHandler(uint16 channel, RequestHandler* rh)
{
	LOCK_RW_MUTEX_WRITE(m_mtxRequestHandlerMutex);

	RequestHandlerListByChannel::iterator pos = m_RequestHandlers.find(channel);
	if (pos != m_RequestHandlers.end())
	{
		RequestHandlerList& handlers = pos->second;
		for (RequestHandlerList::iterator pos = handlers.begin(); pos != handlers.end(); ++pos)
		{
			/*
			if ((*pos)->getHandler() == rh)
			{
				// Disconnect - this will make it safe across copies of the list
				// this is threadsafe and will wait for existing processes to finish
				(*pos)->disconnectHandler();
				handlers.erase(j);	
				break;
			}
			*/
			if (*pos == rh)
			{
				handlers.erase(pos);	
				break;
			}
		}

	}

}
//---------------------------------------------------------------------
void WorkQueue::addResponseHandler(uint16 channel, ResponseHandler* rh)
{
	ResponseHandlerListByChannel::iterator pos = m_ResponseHandlers.find(channel);
	if (pos == m_ResponseHandlers.end())
		pos = m_ResponseHandlers.insert(ResponseHandlerListByChannel::value_type(channel, ResponseHandlerList())).first;

	ResponseHandlerList& handlers = pos->second;
	if (std::find(handlers.begin(), handlers.end(), rh) == handlers.end())
		handlers.push_back(rh);
}
//---------------------------------------------------------------------
void WorkQueue::removeResponseHandler(uint16 channel, ResponseHandler* rh)
{
	ResponseHandlerListByChannel::iterator pos = m_ResponseHandlers.find(channel);
	if (pos != m_ResponseHandlers.end())
	{
		ResponseHandlerList& handlers = pos->second;
		ResponseHandlerList::iterator posTargetHandler = std::find(handlers.begin(), handlers.end(), rh);
		if (posTargetHandler != handlers.end())
			handlers.erase(posTargetHandler);

	}
}
//---------------------------------------------------------------------
WorkQueue::RequestID WorkQueue::addRequest(uint16 channel, uint16 requestType,  const Any& rData, uint16 retryCount, bool forceSynchronous)
{
	Request* req = 0;
	RequestID rid = 0;

	{
		// lock to acquire rid and push request to the queue
		LOCK_MUTEX(m_mtxRequestMutex)

		if (!m_bAcceptRequests || m_bShuttingDown)
			return 0;

		rid = ++m_uRequestCount;
		req = new Request(channel, requestType, rData, retryCount, rid);

		LogManager::getSingleton() << 
			"WorkQueue('" << m_strWorkQueueName << "') - QUEUED(thread:" <<
			THREAD_CURRENT_ID
			<< "): ID=" << rid
			<< " channel=" << channel << " requestType=" << requestType;

		if (!forceSynchronous)
		{
			m_RequestQueue.push_back(req);
			notifyWorkers();
			return rid;
		}
	}


	processRequestResponse(req, true);

	return rid;

}
//---------------------------------------------------------------------
void WorkQueue::addRequestWithRID(WorkQueue::RequestID rid, uint16 channel, uint16 requestType, const Any& rData, uint16 retryCount)
{
	// lock to push request to the queue
	LOCK_MUTEX(m_mtxRequestMutex)

	if (m_bShuttingDown)
		return;

	Request* req = new Request(channel, requestType, rData, retryCount, rid);

	LogManager::getSingleton() << 
		"WorkQueue('" << m_strWorkQueueName << "') - REQUEUED(thread:" <<
		THREAD_CURRENT_ID
		<< "): ID=" << rid
		<< " channel=" << channel << " requestType=" << requestType;

	m_RequestQueue.push_back(req);
	notifyWorkers();

	processRequestResponse(req, true);
}
//---------------------------------------------------------------------
void WorkQueue::abortRequest(RequestID id)
{
	LOCK_MUTEX(m_mtxProcessMutex)

		// NOTE: Pending requests are exist any of RequestQueue, ProcessQueue and
		// ResponseQueue when keeping ProcessMutex, so we check all of these queues.

	for (RequestQueue::iterator i = m_ProcessQueue.begin(); i != m_ProcessQueue.end(); ++i)
	{
		if ((*i)->getID() == id)
		{
			(*i)->abortRequest();
			break;
		}
	}

		{
			LOCK_MUTEX(m_mtxRequestMutex)

				for (RequestQueue::iterator i = m_RequestQueue.begin(); i != m_RequestQueue.end(); ++i)
				{
					if ((*i)->getID() == id)
					{
						(*i)->abortRequest();
						break;
					}
				}
		}

		{
			LOCK_MUTEX(m_mtxResponseMutex)

				for (ResponseQueue::iterator i = m_ResponseQueue.begin(); i != m_ResponseQueue.end(); ++i)
				{
					if( (*i)->getRequest()->getID() == id )
					{
						(*i)->abortRequest();
						break;
					}
				}
		}
}
//---------------------------------------------------------------------
void WorkQueue::abortRequestsByChannel(uint16 channel)
{
	LOCK_MUTEX(m_mtxProcessMutex)

		for (RequestQueue::iterator i = m_ProcessQueue.begin(); i != m_ProcessQueue.end(); ++i)
		{
			if ((*i)->getChannel() == channel)
			{
				(*i)->abortRequest();
			}
		}

		{
			LOCK_MUTEX(m_mtxRequestMutex)

				for (RequestQueue::iterator i = m_RequestQueue.begin(); i != m_RequestQueue.end(); ++i)
				{
					if ((*i)->getChannel() == channel)
					{
						(*i)->abortRequest();
					}
				}
		}

		{
			LOCK_MUTEX(m_mtxResponseMutex)

				for (ResponseQueue::iterator i = m_ResponseQueue.begin(); i != m_ResponseQueue.end(); ++i)
				{
					if( (*i)->getRequest()->getChannel() == channel )
					{
						(*i)->abortRequest();
					}
				}
		}
}
//---------------------------------------------------------------------
void WorkQueue::abortAllRequests()
{
	LOCK_MUTEX(m_mtxProcessMutex)

	for (RequestQueue::iterator i = m_ProcessQueue.begin(); i != m_ProcessQueue.end(); ++i)
	{
		(*i)->abortRequest();
	}

	{
		LOCK_MUTEX(m_mtxRequestMutex)

		for (RequestQueue::iterator i = m_RequestQueue.begin(); i != m_RequestQueue.end(); ++i)
		{
			(*i)->abortRequest();
		}
	}

	{
		LOCK_MUTEX(m_mtxResponseMutex)

		for (ResponseQueue::iterator i = m_ResponseQueue.begin(); i != m_ResponseQueue.end(); ++i)
		{
			(*i)->abortRequest();
		}
	}
}
//---------------------------------------------------------------------
void WorkQueue::setPaused(bool pause)
{
	LOCK_MUTEX(m_mtxRequestMutex)

	m_bPaused = pause;
}
//---------------------------------------------------------------------
bool WorkQueue::isPaused() const
{
	return m_bPaused;
}
//---------------------------------------------------------------------
void WorkQueue::setRequestsAccepted(bool accept)
{
	LOCK_MUTEX(m_mtxRequestMutex)

	m_bAcceptRequests = accept;
}
//---------------------------------------------------------------------
bool WorkQueue::getRequestsAccepted() const
{
	return m_bAcceptRequests;
}
//---------------------------------------------------------------------
void WorkQueue::_processNextRequest()
{
	Request* request = 0;
	{
		// scoped to only lock while retrieving the next request
		LOCK_MUTEX(m_mtxProcessMutex)
		{
			LOCK_MUTEX(m_mtxRequestMutex)

			if (!m_RequestQueue.empty())
			{
				request = m_RequestQueue.front();
				m_RequestQueue.pop_front();
				m_ProcessQueue.push_back( request );
			}
		}
	}

	if (request)
	{
		processRequestResponse(request, false);
	}


}
//---------------------------------------------------------------------
void WorkQueue::processRequestResponse(Request* r, bool synchronous)
{
	Response* response = processRequest(r);

	LOCK_MUTEX(m_mtxProcessMutex)

	RequestQueue::iterator it;
	for( it = m_ProcessQueue.begin(); it != m_ProcessQueue.end(); ++it )
	{
		if( (*it) == r )
		{
			m_ProcessQueue.erase( it );
			break;
		}
	}

	if (response != NULL)
	{
		if (!response->succeeded())
		{
			// Failed, should we retry?
			const Request* req = response->getRequest();
			if (req->getRetryCount())
			{
				addRequestWithRID(req->getID(), req->getChannel(), req->getType(), req->getData(), 
					req->getRetryCount() - 1);
				// discard response (this also deletes request)
				delete response;
				return;
			}
		}
		if (synchronous)
		{
			processResponse(response);
			delete response;
		}
		else
		{
			if( response->getRequest()->getAborted() )
			{
				// destroy response user data
				response->abortRequest();
			}
			// Queue response
			LOCK_MUTEX(m_mtxResponseMutex)
			m_ResponseQueue.push_back(response);
			// no need to wake thread, this is processed by the main thread
		}

	}
	else
	{
		// no response, delete request
		LogManager::getSingleton() << 
			"WorkQueue('" << m_strWorkQueueName << "') warning: no handler processed request "
			<< r->getID() << ", channel " << r->getChannel()
			<< ", type " << r->getType();
		delete r;
	}

}
//---------------------------------------------------------------------
void WorkQueue::processResponses() 
{
	//unsigned long msStart = Root::getSingleton().getTimer()->getMilliseconds();
	unsigned long msStart = 0; 
	unsigned long msCurrent = msStart;

	// keep going until we run out of responses or out of time
	while(true)
	{
		Response* response = 0;
		{
			LOCK_MUTEX(m_mtxResponseMutex)

			if (m_ResponseQueue.empty())
				break; // exit loop
			else
			{
				response = m_ResponseQueue.front();
				m_ResponseQueue.pop_front();
			}
		}

		if ( response != NULL )
		{
			processResponse(response);

			delete response;
		}

		// time limit
		if (m_uResponseTimeLimitMS > 0)
		{
			//msCurrent = Root::getSingleton().getTimer()->getMilliseconds();
			msCurrent = 1;
			if (msCurrent - msStart > m_uResponseTimeLimitMS)
				break;
		}
	}
}
//---------------------------------------------------------------------
WorkQueue::Response* WorkQueue::processRequest(Request* r)
{
	RequestHandlerListByChannel handlerListCopy;
	{
		// lock the list only to make a copy of it, to maximise parallelism
		LOCK_RW_MUTEX_READ(m_mtxRequestHandlerMutex);

		handlerListCopy = m_RequestHandlers;

	}

	Response* response = 0;

	std::stringstream dbgMsg;
	dbgMsg <<
		THREAD_CURRENT_ID
		<< "): ID=" << r->getID() << " channel=" << r->getChannel() 
		<< " requestType=" << r->getType();

	LogManager::getSingleton() << 
		"WorkQueue('" << m_strWorkQueueName << "') - PROCESS_REQUEST_START(" << dbgMsg.str();

	RequestHandlerListByChannel::iterator i = handlerListCopy.find(r->getChannel());
	if (i != handlerListCopy.end())
	{
		RequestHandlerList& handlers = i->second;
		for (RequestHandlerList::reverse_iterator j = handlers.rbegin(); j != handlers.rend(); ++j)
		{
			// threadsafe call which tests canHandleRequest and calls it if so 
			response = (*j)->handleRequest(r, this);

			if (response == NULL)
				break;
		}
	}

	LogManager::getSingleton() << 
		"WorkQueue('" << m_strWorkQueueName << "') - PROCESS_REQUEST_END(" << dbgMsg.str()
		<< " processed=" << (response!=0);

	return response;

}
//---------------------------------------------------------------------
void WorkQueue::processResponse(Response* r)
{
	std::stringstream dbgMsg;
	dbgMsg << "thread:" <<
		THREAD_CURRENT_ID
		<< "): ID=" << r->getRequest()->getID()
		<< " success=" << r->succeeded() << " messages=[" << r->getMessages() << "] channel=" 
		<< r->getRequest()->getChannel() << " requestType=" << r->getRequest()->getType();

	LogManager::getSingleton() << 
		"WorkQueue('" << m_strWorkQueueName << "') - PROCESS_RESPONSE_START(" << dbgMsg.str();

	ResponseHandlerListByChannel::iterator i = m_ResponseHandlers.find(r->getRequest()->getChannel());
	if (i != m_ResponseHandlers.end())
	{
		ResponseHandlerList& handlers = i->second;
		for (ResponseHandlerList::reverse_iterator j = handlers.rbegin(); j != handlers.rend(); ++j)
		{
			if ((*j)->canHandleResponse(r, this))
			{
				(*j)->handleResponse(r, this);
			}
		}
	}
	
	LogManager::getSingleton() << 
		"WorkQueue('" << m_strWorkQueueName << "') - PROCESS_RESPONSE_END(" << dbgMsg.str();
}
//---------------------------------------------------------------------
bool WorkQueue::isShuttingDown() const
{
	return m_bShuttingDown;
}

//---------------------------------------------------------------------

void WorkQueue::WorkerFunc::operator()()
{
	m_pTargetWorkQueue->_threadMain();
}

void WorkQueue::WorkerFunc::run()
{
	m_pTargetWorkQueue->_threadMain();
}
